media_pp\elements\sink/
rtsp_sink.rs

1use std::{ffi::CString, ptr, sync::Arc};
2
3use crate::pp_log::{PpLog, pp_error, pp_info};
4use ffmpeg_next::{self as ffmpeg, ffi};
5use thiserror::Error as ThisError;
6
7use crate::{
8    buffer::MediaBuffer,
9    control::ControlMsg,
10    element::{Element, ElementType, Sink, element_pp_log},
11    elements::RtspTransport,
12    error::Result,
13};
14
15/// Errors produced while opening or writing an [`RtspSink`].
16#[derive(Debug, ThisError)]
17pub enum RtspSinkError {
18    #[error("ffmpeg error: {0}")]
19    Ffmpeg(#[from] ffmpeg::Error),
20
21    #[error(
22        "RtspSink only remuxes compressed Packets, got a decoded {0}; \
23         connect an encoder or demuxer packet pad instead"
24    )]
25    UnsupportedBuffer(&'static str),
26
27    #[error("RTSP URL contains a NUL byte")]
28    InvalidUrl,
29}
30
31/// Publishes one compressed packet stream to an already-running RTSP server.
32///
33/// [`RtspSink::open`] performs the RTSP `ANNOUNCE`/`SETUP`/`RECORD`
34/// handshake through FFmpeg, so the server must already be listening at
35/// `url` and must permit publishing to that path. The server can be
36/// MediaMTX or any other implementation that accepts RTSP publishing;
37/// this element does not start, stop, or otherwise depend on a particular
38/// server process.
39///
40/// This is a remuxing sink, not an encoder. Incoming buffers must be
41/// compressed [`MediaBuffer::Packet`] values whose codec parameters and
42/// time base match the values passed to [`RtspSink::open`]. Place a
43/// [`crate::elements::Pacer`] upstream when publishing packets from a file,
44/// otherwise the file will be sent faster than real time.
45///
46/// The current sink publishes one stream. Build separate sinks and RTSP
47/// paths when publishing independent streams.
48pub struct RtspSink {
49    pp_log: PpLog,
50    name: Arc<str>,
51    url: String,
52    output: ffmpeg::format::context::Output,
53    input_time_base: ffmpeg::Rational,
54    last_output_dts: Option<i64>,
55    last_output_pts: Option<i64>,
56    pts_offset: i64,
57    pending_seek: bool,
58}
59
60impl RtspSink {
61    /// Connects to `url` and starts publishing.
62    ///
63    /// `params` and `time_base` must describe every packet subsequently
64    /// passed to [`Sink::consume`]. TCP is the most reliable transport for
65    /// general networks; UDP is useful when the network path and server
66    /// permit the negotiated RTP/RTCP ports.
67    pub fn open(
68        name: impl Into<String>,
69        url: impl Into<String>,
70        transport: RtspTransport,
71        params: ffmpeg::codec::Parameters,
72        time_base: ffmpeg::Rational,
73    ) -> Result<Self> {
74        let url = url.into();
75        let mut output = alloc_output(&url)?;
76
77        {
78            let mut stream = output
79                .add_stream(ffmpeg::encoder::find(ffmpeg::codec::Id::None))
80                .map_err(RtspSinkError::from)?;
81            stream.set_parameters(params);
82            // Avoid codec-tag incompatibilities when the input packet came
83            // from a container with a different tag convention.
84            unsafe {
85                (*stream.parameters().as_mut_ptr()).codec_tag = 0;
86            }
87            stream.set_time_base(time_base);
88        }
89
90        let mut options = ffmpeg::Dictionary::new();
91        options.set("rtsp_transport", transport.as_ffmpeg_option());
92        output
93            .write_header_with(options)
94            .map_err(RtspSinkError::from)?;
95
96        let name: Arc<str> = name.into().into();
97        let pp_log = element_pp_log(ElementType::RtspSink, &name, None);
98        pp_info!(pp_log: &pp_log, "publishing: url={url}, transport={transport:?}");
99
100        Ok(Self {
101            pp_log,
102            name,
103            url,
104            output,
105            input_time_base: time_base,
106            last_output_dts: None,
107            last_output_pts: None,
108            pts_offset: 0,
109            pending_seek: false,
110        })
111    }
112
113    /// URL this sink publishes to.
114    pub fn url(&self) -> &str {
115        &self.url
116    }
117}
118
119/// Allocates an RTSP muxer without opening a generic `AVIOContext`.
120///
121/// RTSP is a libavformat muxer, not a generic AVIO protocol. Its muxer
122/// owns the control and RTP sockets internally during header/packet writes,
123/// while `ffmpeg_next::format::output_as` attempts an incompatible generic
124/// `avio_open2` first on FFmpeg builds where `rtsp` is not an AVIO protocol.
125fn alloc_output(url: &str) -> Result<ffmpeg::format::context::Output> {
126    let c_url = CString::new(url).map_err(|_| RtspSinkError::InvalidUrl)?;
127    let c_format = CString::new("rtsp").expect("static format name contains no NUL");
128
129    unsafe {
130        let mut context: *mut ffi::AVFormatContext = ptr::null_mut();
131        let result = ffi::avformat_alloc_output_context2(
132            &mut context,
133            ptr::null_mut(),
134            c_format.as_ptr(),
135            c_url.as_ptr(),
136        );
137        if result < 0 {
138            return Err(RtspSinkError::Ffmpeg(ffmpeg::Error::from(result)).into());
139        }
140
141        Ok(ffmpeg::format::context::Output::wrap(context))
142    }
143}
144
145impl Element for RtspSink {
146    fn name(&self) -> Arc<str> {
147        self.name.clone()
148    }
149
150    fn element_type(&self) -> ElementType {
151        ElementType::RtspSink
152    }
153
154    fn pp_log(&self) -> &PpLog {
155        &self.pp_log
156    }
157
158    fn pp_log_mut(&mut self) -> &mut PpLog {
159        &mut self.pp_log
160    }
161}
162
163impl Sink for RtspSink {
164    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
165        match buf {
166            MediaBuffer::Packet(packet) => {
167                let mut packet = (*packet).clone();
168                let output_time_base = self
169                    .output
170                    .stream(0)
171                    .expect("stream 0 was added by RtspSink::open")
172                    .time_base();
173                packet.rescale_ts(self.input_time_base, output_time_base);
174
175                if let Some(raw_pts) = packet.pts() {
176                    if self.pending_seek {
177                        // Keep the published timeline monotonic across an
178                        // upstream seek. DTS is the muxer's hard ordering
179                        // requirement; PTS is the fallback for packets that
180                        // carry no DTS.
181                        self.pts_offset = match (self.last_output_dts, packet.dts()) {
182                            (Some(last_dts), Some(raw_dts)) => last_dts + 1 - raw_dts,
183                            _ => match self.last_output_pts {
184                                Some(last_pts) => last_pts + 1 - raw_pts,
185                                None => 0,
186                            },
187                        };
188                        self.pending_seek = false;
189                    }
190
191                    let corrected_pts = raw_pts + self.pts_offset;
192                    packet.set_pts(Some(corrected_pts));
193                    if let Some(raw_dts) = packet.dts() {
194                        let corrected_dts = raw_dts + self.pts_offset;
195                        packet.set_dts(Some(corrected_dts));
196                        self.last_output_dts = Some(corrected_dts);
197                    }
198                    self.last_output_pts = Some(corrected_pts);
199                }
200
201                packet.set_stream(0);
202                packet.set_position(-1);
203                packet
204                    .write_interleaved(&mut self.output)
205                    .map_err(RtspSinkError::from)
206                    .map_err(Into::into)
207                    .inspect_err(|error| pp_error!(self, "write_interleaved failed: {error}"))
208            }
209            MediaBuffer::Eos => self
210                .output
211                .write_trailer()
212                .map_err(RtspSinkError::from)
213                .map_err(Into::into)
214                .inspect_err(|error| pp_error!(self, "write_trailer failed: {error}")),
215            MediaBuffer::Video(_) => {
216                pp_error!(self, "unsupported buffer: Video");
217                Err(RtspSinkError::UnsupportedBuffer("Video").into())
218            }
219            MediaBuffer::Audio(_) => {
220                pp_error!(self, "unsupported buffer: Audio");
221                Err(RtspSinkError::UnsupportedBuffer("Audio").into())
222            }
223        }
224    }
225
226    fn control(&mut self, msg: ControlMsg) -> Result<()> {
227        match msg {
228            ControlMsg::Seek(_) => self.pending_seek = true,
229            ControlMsg::Pause | ControlMsg::Resume | ControlMsg::Stop => {}
230        }
231        Ok(())
232    }
233}
234
235impl Drop for RtspSink {
236    fn drop(&mut self) {
237        pp_info!(
238            self,
239            "dropped: closing publisher connection to {}",
240            self.url
241        );
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use ffmpeg_next as ffmpeg;
248
249    use super::{RtspSink, RtspSinkError};
250    use crate::{elements::RtspTransport, error::Error};
251
252    #[test]
253    fn rejects_a_url_containing_a_nul_byte_before_connecting() {
254        let result = RtspSink::open(
255            "rtsp",
256            "rtsp://127.0.0.1:8554/stream\0invalid",
257            RtspTransport::Tcp,
258            ffmpeg::codec::Parameters::new(),
259            ffmpeg::Rational(1, 1_000),
260        );
261
262        assert!(matches!(
263            result,
264            Err(Error::RtspSinkError(RtspSinkError::InvalidUrl))
265        ));
266    }
267}
media_pp\elements\sink/rtsp_sink.rs

media_pp\elements\sink/
rtsp_sink.rs
